package cn.easy.rpc.io;

import cn.easy.rpc.core.DataExchangeIdentity;
import cn.easy.rpc.core.RegistrationTable;
import cn.easy.rpc.config.ConfigHolder;
import cn.easy.rpc.core.ServiceRegistrar;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.context.ApplicationListener;
import org.springframework.context.event.ContextRefreshedEvent;
import org.springframework.stereotype.Component;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * @author zwb
 * @create 2020-09-06 14:00
 **/
@Component
public class RegistrarServer implements ApplicationListener<ContextRefreshedEvent> {


    public static volatile boolean off = false;

    private static final Log logger = LogFactory.getLog(RegistrarServer.class);
    //最大核心线程数是瓶颈啊
    private static final ThreadPoolExecutor worker = new ThreadPoolExecutor(3, 8, 15L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(32), r -> {
        Thread thread = new Thread(r);
        thread.setName("registrar worker");
        return thread;
    });

    @Override
    public void onApplicationEvent(ContextRefreshedEvent contextStartedEvent) {
        CompletableFuture.runAsync(() -> {
            ServerSocket serverSocket = null;
            try {
                serverSocket = new ServerSocket(ConfigHolder.serviceCenterPort);
            } catch (IOException e) {
                e.printStackTrace();
            }
            while (!off) {
                Socket accept = null;
                try {
                    if ((accept = serverSocket.accept()) != null) {
                        logger.info("Found a new service node,from " + accept.getInetAddress());
                        worker.submit(new ClientHandler(accept));
                    }
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        });
        logger.info("registrar server started at port " + ConfigHolder.serviceCenterPort);
    }


    class ClientHandler implements Runnable {
        protected volatile boolean isLostContact = false;
        protected AtomicInteger heartBeatCountDown;

        private final Socket socket;
        private String serviceId;

        HeartBeatMonitor monitor;

        public ClientHandler(Socket socket) {
            this.socket = socket;
            heartBeatCountDown = new AtomicInteger(3);
            monitor = new HeartBeatMonitor();
        }

        @Override
        public void run() {
            ObjectInputStream is = null;
            ObjectOutputStream os = null;
            try {
                while (!isLostContact) {
                    is = new ObjectInputStream(new BufferedInputStream(
                            socket.getInputStream()));
                    DataExchangeIdentity req = (DataExchangeIdentity) is.readObject();
                    //初始服务
                    if(req.getPing() != null) {
                        setHost(req.getPing(), socket);
                        RegistrationTable.register(req.getPing());
                        this.serviceId = req.getPing().getServiceId();
                    }
                    heartBeatCountDown.set(3);
                    //响应注册表的信息
                    os = new ObjectOutputStream(socket.getOutputStream());
                    os.writeObject(new DataExchangeIdentity(null, new CopyOnWriteArrayList<>(RegistrationTable.getRegisterTable().stream().filter(registrar -> !registrar.getServiceId().equals(serviceId)).collect(Collectors.toList()))));
                }
            } catch (Exception ex) {
                //ignore
            } finally {
                try {
                    if(is != null)
                    is.close();
                    if(os != null)
                    os.close();
                    socket.close();
                } catch (Exception ex) {
                    ex.printStackTrace();
                }
            }
        }

        class HeartBeatMonitor {
            ScheduledExecutorService executorService;

            HeartBeatMonitor() {
                executorService = Executors.newScheduledThreadPool(1);
                inspect();
            }

            private void inspect() {
                executorService.scheduleWithFixedDelay(() -> {
                    if (heartBeatCountDown.decrementAndGet() <= 0) {
                        //主观下线
                        isLostContact = true;
                        RegistrationTable.offline(serviceId);
                        destroy();
                    }
                }, 3000L, 3000L, TimeUnit.MILLISECONDS);
            }

            void destroy() {
                executorService.shutdownNow();
            }
        }
    }

    private void setHost(ServiceRegistrar serviceRegistrar, Socket socket) {
        serviceRegistrar.setHost(socket.getInetAddress().getHostAddress());
    }
}
